/*
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or (at
 * your option) any later version.
 *
 * This program is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
 * 02110-1301, USA.
 * */

// AmzServer - Tototoy's server model.
// Author: tonyjobmails@gmail.com (tonyhack).
//
// Define class RingQueueSingleModelBoundless.
//
// RingQueueSingleModelBoundless is an implement of RingQueueInterface. This
// implement is a version of "double-thread safe". That is to say, there are only 2
// threads in this producer/consumer model, the producer one only read from the
// queue, and the consumer one only write to it, RingQueueSingleModelBoundless
// is thread-safe in this condition. Because of using lock-free, This RingQueueSin-
// gleModelBoundless performances a high efficiency in this model.
// This RingQueueSingleModelBoundless is boundless, it uses ByteBufferDynamic
// as it's extended queue. The extended queue is a extendable buffer, it can reduce
// the times of memory allocation, so it makes RingQueueSingleModelBoundless
// more efficientive.
// RingQueueSingleModelBoundless also supports blocking pattern, that is to say,
// when the consumer read from the queue, if the queue is empty, the consumer
// will enter into the waiting state, it will be blocked until the producer thread writes
// any new data into the queue. however, RingQueueSingleModelBoundless also
// supports non-blocking pattern.
// Attention, the producer thread does not permit to quit until the extended queue
// is empty. the poducer thread can invokes Shift2Ring(), if the return value is true,
// it can quit immediately, and if not, the producer thread should invoke Shift2Ring()
// after serval milliseconds or longer, this invoking frequency depend on the speed
// of consumer thread.
//

#include <string.h>

#include "ring_queue_single_model_boundless.h"
#include "logging.h"
#include "auto_mutex.h"

namespace amz {

// ring_(ring_size + 1) will call the constructor with giving the argument ring_size + 1,
// this "ring_size + 1" gives the real size of the ring queue, because of being the ring
// queue, there is a useless byte to make the queue to be a ring.
RingQueueSingleModelBoundless::RingQueueSingleModelBoundless(uint32 ring_size,
       uint32 dynamic_queue_extend_size, bool blocking) :
       ring_(ring_size + 1),
       ring_size_(ring_size),
       write_pointer_(0),
       read_pointer_(0),
       dynamic_queue_(dynamic_queue_extend_size),
       blocking_(blocking),
       produce_completely_(false) {
  if(ring_size == 0) {
      LOG(ERROR) << "Error size allocate in RingQueueSingleModelBoundless.";
  }
}

RingQueueSingleModelBoundless::~RingQueueSingleModelBoundless() {}

int RingQueueSingleModelBoundless::Add(const char* element, uint32 size) {
  // First shift data elements to ring queue as much as possible, second
  // try to add element to ring queue.
  if(this->Shift2Ring() == false || this->TryAdd(element, size) == false) {
    // If the space is not enough, check and allocate the dynamic queue
    // and add to dynamic queue.
    uint32 write_size = size + sizeof(RingQueueElement);
    if(this->dynamic_queue_.PreWrite(write_size) == false) {
      LOG(ERROR) << "Failed to Add, dynamic_queue_ size is not enough.";
      return -1;
    }
    RingQueueElement *temp_element
           = (RingQueueElement *)(this->dynamic_queue_.WritePointer());
    temp_element->data_size_ = size;
    memcpy(temp_element->data_pointer_, element, size);
    // Offset data and pointers.
    this->dynamic_queue_.WriteShift(write_size);
  }
  
  return size;
}

int RingQueueSingleModelBoundless::Remove(char* element, uint32 max_size) {
  // Save the write pointer, in case producer thread altering.
  uint32 write_pointer = this->write_pointer_;
  uint32 can_read_size = 0;

  RING_QUEUE_READ_SIZE(this->ring_size_, this->read_pointer_,
         write_pointer, can_read_size);

  // Empty.
  while(can_read_size == 0) {
    // blocking pattern.
    if(this->blocking_ == true) {
      // if producer complete, return 0 and notify complete.
      if(this->CheckAllProducersComplete()) {
        return 0; 
      }
      AutoMutex auto_mutex(&this->mutex_);
      this->condition_not_empty_.Wait(&this->mutex_);
      write_pointer = this->write_pointer_;
      RING_QUEUE_READ_SIZE(this->ring_size_, this->read_pointer_,
             write_pointer, can_read_size);
    } else {
      return 0;
    }
  }
  
  if(can_read_size < sizeof(RingQueueElement)) {
    LOG(ERROR) << "Error, read size: " << can_read_size;
    return -1;
  }
  
  // Get real pointer of ring.
  char *ring_pointer = (char *)this->ring_.BufferPointer();
	
  // Get the read size after read_pointer_.
  uint32 after_read_pointer_size = 0;
  RING_QUEUE_READ_SIZE_AFTER_READ_POINTER(
        this->ring_size_, this->read_pointer_,
        write_pointer, after_read_pointer_size);

		 
  if(after_read_pointer_size >= sizeof(RingQueueElement)) {
    RingQueueElement *temp_element
           = (RingQueueElement *)(&(ring_pointer[this->read_pointer_]));
    uint32 read_size = temp_element->data_size_ + sizeof(RingQueueElement);
    // Check valid
    if(read_size > can_read_size || temp_element->data_size_ > max_size) {
      LOG(ERROR) << "Error size, read_size="
	         << read_size
                 << ", can_read_size="
                 << can_read_size
                 << ", max_size="
                 << max_size;
      return -1;
    }
    // Read data.
    if(after_read_pointer_size >= read_size) {
      // All this element data after read_pointer_.
      memcpy(element, temp_element->data_pointer_, temp_element->data_size_);
      this->read_pointer_ += read_size;
    } else if(after_read_pointer_size == read_size) {
      memcpy(element, temp_element->data_pointer_, temp_element->data_size_);
      // If read to the end, set read_pointer_ 0 value.
      this->read_pointer_ = 0;
    } else {
      // Copy the data from ring queue, when copy the first part of data, we should
      // filter the head size.
      uint32 first_part_size = after_read_pointer_size - sizeof(RingQueueElement);
      // first part.
      memcpy(element, temp_element->data_pointer_, first_part_size);
      uint32 second_part_size = read_size - after_read_pointer_size;
      // second part.
      memcpy(&element[first_part_size], ring_pointer, second_part_size);
      this->read_pointer_ = second_part_size;
    }

    return temp_element->data_size_;
	
  } else {
    // The read size after read_pointer is less than an head size.
    RingQueueElement temp_element;
    char* head = (char *)(&temp_element);
    uint32 second_part_size = sizeof(RingQueueElement) - after_read_pointer_size;
    // Divide into 2 parts to copy a head.
    memcpy(head, &(ring_pointer[this->read_pointer_]), after_read_pointer_size);
    memcpy(&head[after_read_pointer_size], ring_pointer, second_part_size);
    if(can_read_size < temp_element.data_size_ + sizeof(RingQueueElement)) {
      LOG(ERROR) << "Error size, read size is too small, can_read_size="
                 << can_read_size
                 << ", need size="
                 << temp_element.data_size_ + sizeof(RingQueueElement);
	  return -1;
    }
    // Copy the data directly.
    memcpy(element, &ring_pointer[second_part_size], temp_element.data_size_);
    this->read_pointer_ = temp_element.data_size_ + second_part_size;

    return temp_element.data_size_;
  }
  
  return 0;
}

bool RingQueueSingleModelBoundless::Flush() {
  // Shift data elements to ring_ actively.
  return this->Shift2Ring();
}

void RingQueueSingleModelBoundless::ProduceCompletely() {
  // Blocking pattern.
  if(this->blocking_ == true) {
    AutoMutex auto_mutex(&this->mutex_);
    this->condition_not_empty_.Signal();
  }

  this->produce_completely_ = true;
}

bool RingQueueSingleModelBoundless::CheckAllProducersComplete() {
  return this->produce_completely_;
}

void RingQueueSingleModelBoundless::LogPrint() {
  LOG(INFO) << "write_pointer_="
            << this->write_pointer_
            << " read_pointer_="
            << this->read_pointer_
            << " ring_size_="
            << this->ring_size_;
}

bool RingQueueSingleModelBoundless::Shift2Ring() {
  RingQueueElement *temp_element = NULL;
  while(this->dynamic_queue_.ReadSize() != 0) {
    temp_element = (RingQueueElement *)(this->dynamic_queue_.ReadPointer());

    if(this->TryAdd(temp_element->data_pointer_, temp_element->data_size_) == false) {
      return false;
    }
    // TryAdd success, shift from dynamic_queue_.
    this->dynamic_queue_.ReadShift(temp_element->data_size_ + sizeof(RingQueueElement));
  }
  // Go here, means shift all data elements into the ring_queue.
  return true;
}

bool RingQueueSingleModelBoundless::TryAdd(const char *element, uint32 size) {
  // Save the read pointer, in case consumer thread altering.
  uint32 read_pointer = this->read_pointer_;
  uint32 need_size = sizeof(RingQueueElement) + size;
  uint32 free_size = 0;
  
  // Get the free size.
  RING_QUEUE_FREE_SIZE(this->ring_size_, read_pointer,
         this->write_pointer_, free_size);
  if(free_size < need_size) {
    return false;
  }
  
  // Get the real pointer of ring.
  char *ring_pointer = (char *)this->ring_.BufferPointer();
  // Get the free size after write_pointer_.
  uint32 after_write_pointer_size = 0;
  RING_QUEUE_FREE_SIZE_AFTER_WRITE_POINTER(
         this->ring_size_, read_pointer,
         this->write_pointer_, after_write_pointer_size);
	
  if(after_write_pointer_size >= need_size) {
    // Give the size of data at the begin position.
    RingQueueElement *temp_element
	       = (RingQueueElement *)(&(ring_pointer[this->write_pointer_]));
    temp_element->data_size_ = size;
    memcpy(temp_element->data_pointer_, element, size);
    // This position write_pointer_'s altering must behind the copying.
    this->write_pointer_ += need_size;
  } else {
    // Not enough, copy to the end position and copy the left to the begin position.
    if(after_write_pointer_size >= sizeof(RingQueueElement)) {
      // The size after the write position have enough space to hold the size of element.
      RingQueueElement *temp_element =
             (RingQueueElement *)(&(ring_pointer[this->write_pointer_]));
      temp_element->data_size_ = size;
      after_write_pointer_size -= sizeof(RingQueueElement);
      memcpy(temp_element->data_pointer_, element, after_write_pointer_size);
      uint32 second_part_size = size - after_write_pointer_size;
      memcpy(ring_pointer, &element[after_write_pointer_size], second_part_size);
      this->write_pointer_ = second_part_size;
    } else {
      // Not enough, copy the first part of "size" to the write position, then copy the
      // second part to the begin position, at last, copy the element.
      memcpy(&(ring_pointer[this->write_pointer_]),
             (char *)(&size), after_write_pointer_size);
      uint32 second_part_size = sizeof(RingQueueElement) - after_write_pointer_size;
      memcpy(ring_pointer, &(((char *)(&size))[after_write_pointer_size]),
             second_part_size);
      memcpy(&ring_pointer[second_part_size], element, size);
      // This position write_pointer_'s altering must behind the copying action.
      this->write_pointer_ = need_size - after_write_pointer_size;
    }
  }
  
  // Blocking pattern.
  if(this->blocking_ == true) {
    AutoMutex auto_mutex(&this->mutex_);
    this->condition_not_empty_.Signal();
  }

  return true;
}

}  // namespace amz

